package com.avo.ramcache.persist.timing;

import com.avo.ramcache.config.IEntity;
import com.avo.ramcache.orm.Accessor;
import com.avo.ramcache.persist.Element;
import com.avo.ramcache.persist.Listener;
import com.avo.ramcache.util.DateUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collection;
import java.util.concurrent.atomic.AtomicInteger;

public class TimingConsumer implements Runnable {

    private static final Logger LOGGER = LoggerFactory.getLogger(TimingConsumer.class);

    private final String name;
    private final String cron;
    private final Accessor accessor;
    private final TimingPersister owner;
    private final Thread me;
    private TimingConsumerState state;
    private volatile boolean stoped;
    private long nextTime;
    private final AtomicInteger error = new AtomicInteger();
    private volatile boolean interrupted;

    public TimingConsumer(String name, String cron, Accessor accessor, TimingPersister owner) {
        this.name = name;
        this.cron = cron;
        this.owner = owner;
        this.accessor = accessor;
        this.me = new Thread(this, "持久化" + name + "定时");
        this.me.setDaemon(true);
        this.me.start();
    }

    @Override
    public void run() {
        while (!this.stoped) {
            try {
                Collection<Element> elements = null;
                synchronized (this.me) {
                    this.state = TimingConsumerState.WAITING;
                    if (!this.interrupted) {
                        this.nextTime = DateUtils.getNextTime(cron);
                        // log
                        try {
                            long ms = this.nextTime - System.currentTimeMillis();
                            if (ms > 0) {
                                this.me.wait(ms);
                            }
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }

                    elements = this.owner.clearElements();
                    this.interrupted = false;
                    this.state = TimingConsumerState.RUNNING;
                }

//                Date start = new Date();
                // log
                // 持久化
                persist(elements);
                // log
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        synchronized (me) {
            Collection<Element> elements = owner.clearElements();
            persist(elements);
            state = TimingConsumerState.STOPPED;
            interrupted = false;
        }
    }

    private void persist(Collection<Element> elements) {
        for (Element element : elements) {
            boolean succFlag = false;
            IEntity<?> entity = element.getEntity();
            Listener listener;
            try {
                Class clz = element.getEntityClass();
                switch (element.getType()) {
                    case SAVE:
                        if (element.getEntity().serialize()) {
                            this.accessor.save(clz, element.getEntity());
                            succFlag = true;
                        }
                        break;
                    case UPDATE:
                        if (element.getEntity().serialize()) {
                            this.accessor.saveOrUpdate(clz, element.getEntity());
                            succFlag = true;
                        }
                        break;
                    case REMOVE:
                        this.accessor.remove(clz, element.getId());
                        succFlag = true;
                        break;
                }

                listener = this.owner.getListener(clz);
                if (listener != null) {
                    listener.notify(element.getType(), succFlag, element.getId(), element.getEntity(), null);
                }

            } catch (RuntimeException e) {
                this.error.getAndIncrement();
                LOGGER.error("实体更新队列{" + name + "}处理元素{" + element + "}时data{" + entity + "}出现异常", e);
                listener = this.owner.getListener(element.getEntityClass());
                if (listener != null) {
                    listener.notify(element.getType(), succFlag, element.getId(), element.getEntity(), e);
                }
            } catch (Exception e) {
                this.error.getAndIncrement();
                LOGGER.error("更新队列出现未知异常", e);
            }
        }
    }

    public void persistNow(Collection<Element> elements) {
        synchronized (me) {
            persist(elements);
        }
    }

    public void stop() {
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("定时入库[{}]收到停止通知", name);
        }
        synchronized (me) {
            stoped = true;
            interrupted = true;
            me.notify();
        }
        while (interrupted) {
            Thread.yield();
        }
    }

    public void interrupt() {
        synchronized (me) {
            interrupted = true;
            me.notify();
        }
        while (interrupted) {
            Thread.yield();
        }
    }

    public TimingConsumerState getState() {
        synchronized (me) {
            return state;
        }
    }

    public long getNextTime() {
        return nextTime;
    }
}
